package com.hjs.mr.merge_and_sort;

import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.io.IOUtils;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.mapreduce.RecordWriter;
import org.apache.hadoop.mapreduce.TaskAttemptContext;

import java.io.IOException;

/**
 * 自定义的RecordWriter，定义了文件的输出方式
 */
public class SortRecordWriter extends RecordWriter<SortBean, NullWritable> {

    FSDataOutputStream outputStream = null;

    public SortRecordWriter(FSDataOutputStream outputStream) {
        this.outputStream = outputStream;
    }

    /**
     * 定义了文件的输出方式和格式，在这里可以使用自定义的方法使用不同的输出流将数据输出到不同的文件中，但是此次的任务并没有类似的相关需求，所以在此没有进行相关的设定
     * @param sortBean
     * @param nullWritable
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void write(SortBean sortBean, NullWritable nullWritable) throws IOException, InterruptedException {
        String bean = sortBean.toString();
        outputStream.write(bean.getBytes());
        outputStream.write("\r\n".getBytes());
        outputStream.flush();
    }

    /**
     * 完成输出，关闭输出流
     * @param taskAttemptContext
     * @throws IOException
     * @throws InterruptedException
     */
    @Override
    public void close(TaskAttemptContext taskAttemptContext) throws IOException, InterruptedException {
        IOUtils.closeStream(outputStream);
    }
}
